package com.shanzhu.beadhouse.service.kafka;

import com.shanzhu.beadhouse.common.constant.ConsumeEnum;
import com.shanzhu.beadhouse.message.OrderConsumeMessage;
import com.shanzhu.beadhouse.message.OrderPaymentMessage;
import com.shanzhu.beadhouse.message.OrderTopics;
import com.shanzhu.beadhouse.service.common.ConsumeFunc;
import com.shanzhu.beadhouse.service.common.ElderFunc;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.Date;
import java.util.UUID;

/**
 * 订单支付消息消费者
 */
@Slf4j
@Component
public class OrderPaymentConsumer {

    @Resource
    private ElderFunc elderFunc;

    @Resource
    private ConsumeFunc consumeFunc;

    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * 处理订单支付消息
     */
    @KafkaListener(topics = OrderTopics.ORDER_PAYMENT, groupId = "order-payment-group")
    @Transactional
    public void handleOrderPayment(OrderPaymentMessage message,
                                   Acknowledgment acknowledgment,
                                   @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                                   @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                                   @Header(KafkaHeaders.OFFSET) long offset) {
        log.info("收到订单支付消息: topic={}, partition={}, offset={}, message={}",
                topic, partition, offset, message);

        try {
            // 1. 老人扣费
            elderFunc.deductionFee(message.getElderId(), message.getPayAmount());
            log.info("老人扣费成功: elderId={}, amount={}", message.getElderId(), message.getPayAmount());

            // 2. 新增消费记录
            consumeFunc.addConsume(
                    message.getElderId(),
                    ConsumeEnum.DISHES.getType(),
                    message.getPayAmount(),
                    message.getConsumeDate()
            );
            log.info("消费记录创建成功: elderId={}, type={}, amount={}",
                    message.getElderId(), ConsumeEnum.DISHES.getType(), message.getPayAmount());

            // 3. 发送消费记录消息（可选，用于后续扩展）
            OrderConsumeMessage consumeMessage = new OrderConsumeMessage();
            consumeMessage.setElderId(message.getElderId());
            consumeMessage.setConsumeType(ConsumeEnum.DISHES.getType());
            consumeMessage.setConsumeAmount(message.getPayAmount());
            consumeMessage.setConsumeDate(message.getConsumeDate());
            consumeMessage.setMessageId(UUID.randomUUID().toString());
            consumeMessage.setCreateTime(new Date());

            kafkaTemplate.send(OrderTopics.ORDER_CONSUME, consumeMessage);
            log.info("消费记录消息发送成功: elderId={}, type={}",
                    message.getElderId(), ConsumeEnum.DISHES.getType());

            // 手动确认消息
            acknowledgment.acknowledge();
            log.info("订单支付消息处理完成: orderId={}", message.getOrderId());

        } catch (Exception e) {
            log.error("处理订单支付消息失败: orderId={}, elderId={}",
                    message.getOrderId(), message.getElderId(), e);
            // 抛出异常，消息会重试
            throw e;
        }
    }
}